package com.hliushi.spark.sql

import org.apache.spark.sql.SparkSession
import org.junit.Test

/**
 * descriptions:
 *
 * author: Hliushi
 * date: 2021/6/11 8:00
 */
class Practice {

  // 1.创建SparkSession对象
  val spark: SparkSession = SparkSession.builder()
    .master("local[6]")
    .appName("mysql write")
    .getOrCreate()

  //import spark.implicits._

  @Test
  def flowProcess(): Unit = {
    val dataFrame = spark.read
      .option("inferSchema", value = true)
      .option("delimiter", "\t")
      .csv("dataset/data_flow.dat")


    dataFrame.printSchema()
    dataFrame.show()

    //val value = dataFrame.map((row: Row) => FlowBean(row.getInt(6), row.getInt(7), row.getInt(8), row.getInt(9)))
    //val value = dataFrame.map((row: Row) => FlowBean(row.getAs[Int](6), row.getAs[Int](7), row.getAs[Int](8), row.getAs[Int](9)))

    // Error:(33, 17) recursive value dataFrame needs type
    //    val value = dataFrame.map((row: Row) => FlowBean(row.getInt(6), row.getInt(7), row.getInt(8), row.getInt(9)))
    //value.show()

  }


  @Test
  def rddFlowProcess(): Unit = {
    val source = spark.sparkContext.textFile("dataset/data_flow.dat")

    val value = source.map((line: String) => {
      val arr = line.split("\t")
      FlowBean(arr(1), arr(6).toInt, arr(7).toInt, arr(8).toInt, arr(9).toInt)
    })

    val result = value.groupBy((flow: FlowBean) => flow.iphone)
      .map((item: (String, Iterable[FlowBean])) => {
        var upFlow = 0
        var downFlow = 0
        var upCountFlow = 0
        var downCountFlow = 0
        item._2.foreach((x: FlowBean) => {
          upFlow += x.upFlow
          downFlow += x.downFlow
          upCountFlow += x.upCountFlow
          downCountFlow += x.downCountFlow
        })
        FlowBean(item._1, upFlow, downFlow, upCountFlow, downCountFlow)
      })


    /**
     * 发现问题, 这个sortBy, 在没有collect的时候并不会触发这个算子
     */
    result
      .sortBy((flow: FlowBean) => flow.upCountFlow, ascending = false)
      .collect()
      .foreach(println)

  }
}

case class FlowBean(iphone: String, var upFlow: Int, var downFlow: Int, var upCountFlow: Int, var downCountFlow: Int)